/**
 * 
 */
package com.db.stockexchange.server.connection;

import java.io.IOException;
import java.net.Socket;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Logger;

import com.db.stockexchange.common.messages.FIXMessage;
import com.db.stockexchange.common.messages.server.PoisonMessage;

/**
 * @author Rayman
 * Represents map of client connections
 */
public class ConnectionPool implements Runnable {
	static final int SLEEP_TIMEOUT = 500;
	
	HashMap<String, Connection> pool = new HashMap<String, Connection>();
	BlockingQueue<FIXMessage> toEngineQueue;
	BlockingQueue<FIXMessage> fromEngineQueue;
	Logger logger;
	
	volatile boolean working = true;
	
	public ConnectionPool(BlockingQueue<FIXMessage> toEngine, BlockingQueue<FIXMessage> fromEngine) {
		toEngineQueue = toEngine;
		fromEngineQueue = fromEngine;
		logger = Logger.getLogger(this.getClass().getName());
		
		new Thread(this, "ConnectionPool").start();
	}
	
	public boolean add(Connection connection) {
		synchronized (pool) {
			if (pool.containsKey(connection.getClientID()))
				return false;
			
			pool.put(connection.getClientID(), connection);
			logger.fine(String.format("Connection for client <%s> added to pool", connection.getClientID()));
			return true;
		}
	}
	
	public void remove(Connection connection) {
		synchronized (pool) {
			if (pool.containsKey(connection.getClientID())) {
				pool.remove(connection.getClientID());
				logger.fine(String.format("Connection for client <%s> removed from pool", connection.getClientID()));
			}
		}
	}
	
	public boolean add(Socket socket) throws IOException {
		Connection connection = new Connection(socket, toEngineQueue, this);
		boolean res = true;//this.add(connection);
		new Thread(connection).start();	
		
		return res;
	}
	
	private void sendMessage(String clientID, FIXMessage message) {
		Connection c = pool.get(clientID);
		if (c != null) {
			c.sendMessage(message);
		}
		else
			logger.warning(String.format("Could not find connection for client <%s>", clientID));		
	}
	
	public void run() {
		// processing messages from engine to clients
		while (working) {
			FIXMessage message = null;
			try {
				message = fromEngineQueue.take();
			} catch (InterruptedException e) { }
			if (message != null && message.getClass() != PoisonMessage.class)
				sendMessage(message.getSenderCompID(), message);
		}
		
		dispose();
	}

	public void close() {
		working = false;
		fromEngineQueue.add(new PoisonMessage());
	}
	
	private void dispose() {
		for (Connection c : pool.values())
			c.close();
		
		pool.clear();
		fromEngineQueue.clear();
	}
}
